package com.lzq.mq.client.producer;

import com.alibaba.fastjson2.JSON;
import com.lzq.mq.common.coder.TcpMsg;
import com.lzq.mq.common.constants.NameServerConstants;
import com.lzq.mq.common.dto.*;
import com.lzq.mq.common.enums.*;
import com.lzq.mq.common.remote.BrokerNettyRemoteClient;
import com.lzq.mq.common.remote.NameServerNettyRemoteClient;
import com.lzq.mq.common.utils.AssertUtils;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 连接nameserver，给nameserver发送心跳，拉取broker地址
 * 与broker建立连接，发送数据给到broker节点
 */
@Slf4j
public class DefaultProducer implements Producer {

    /**
     * nameserver的ip,端口,用户名,密码
     */
    @Getter
    @Setter
    private String nsIp;

    @Getter
    @Setter
    private Integer nsPort;

    @Getter
    @Setter
    private String nsUser;

    @Getter
    @Setter
    private String nsPwd;

    @Getter
    @Setter
    private String brokerGroup;

    @Getter
    @Setter
    private String brokerMode;

    @Getter
    private NameServerNettyRemoteClient nameServerNettyRemoteClient;

    /**
     * broker节点
     */
    @Getter
    @Setter
    private List<String> brokerAddressList;

    /**
     * broker节点和连接客户端工具
     */
    @Getter
    private Map<String,BrokerNettyRemoteClient> brokerNettyRemoteClientMap = new ConcurrentHashMap<>();

    public void start() {

        nameServerNettyRemoteClient = new NameServerNettyRemoteClient(nsIp, nsPort);
        nameServerNettyRemoteClient.buildConnection(); // 建立连接
        // 注册服务
        if (!doRegistry()) {
            log.error("注册服务失败");
            return;
        }
        // 开启心跳
        startHeartBeatTask();
        // 拉取broker节点
        this.fetchBrokerAddress();
        // 连接到broker节点
        this.connectBroker();
    }

    /**
     * 在nameserver注册producer, 使其能够拉取服务
     * @return true:注册成功
     */
    private boolean doRegistry() {
        String registryMsgId = UUID.randomUUID().toString();
        ServiceRegistryReqDTO serviceRegistryReqDTO = new ServiceRegistryReqDTO();
        serviceRegistryReqDTO.setMsgId(registryMsgId);
        serviceRegistryReqDTO.setUser(nsUser);
        serviceRegistryReqDTO.setPassword(nsPwd);
        serviceRegistryReqDTO.setRegistryType(RegistryTypeEnum.PRODUCER.getCode());
        // 设置属性
        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.REGISTRY.getCode(), JSON.toJSONBytes(serviceRegistryReqDTO));
        // 发送注册消息, 并同步等待
        TcpMsg registryResponse = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, registryMsgId);
        if (NameServerResponseCode.REGISTRY_SERVICE_SUCCESS.getCode() == registryResponse.getCode()) {
            return true;
        } else {
            log.info("注册服务失败");
            return false;
        }
    }

    /**
     * 开启心跳任务
     */
    private void startHeartBeatTask() {
        Thread heartBeatTask = new Thread(new Runnable() {
            @Override
            public void run() {
                log.info("开启心跳任务");
                while (true) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(NameServerConstants.SERVER_HEART_BEAT_INTERVAL_MS);
                        String heartBeatMsgId = UUID.randomUUID().toString();
                        HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
                        heartBeatDTO.setMsgId(heartBeatMsgId);
                        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.HEART_BEAT.getCode(),
                                JSON.toJSONBytes(heartBeatDTO));
                        TcpMsg heartBeatResponse = nameServerNettyRemoteClient.sendSyncMsg(tcpMsg, heartBeatMsgId);
                    } catch (InterruptedException e) {
                        log.error("心跳任务异常", e);
                    }
                }
            }
        }, "DefaultProducer心跳任务");
        heartBeatTask.start();
    }

    /**
     * 拉取broker节点ip地址
     */
    public void fetchBrokerAddress() {

        String fetchBrokerAddressMsgId = UUID.randomUUID().toString();
        PullBrokerIpReqDTO pullBrokerIpDTO = new PullBrokerIpReqDTO();
        // 单机模式拉取single, 其他的拉取 master
        if (brokerMode == null || brokerMode.isEmpty() || BrokerClusterModeEnum.SINGLE.getCode().equals(brokerMode)) {
            pullBrokerIpDTO.setRole(BrokerRegistryRoleEnum.SINGLE.getCode());
        } else {
            pullBrokerIpDTO.setRole(BrokerRegistryRoleEnum.MASTER.getCode());
        }

        pullBrokerIpDTO.setMsgId(fetchBrokerAddressMsgId);
        TcpMsg req = new TcpMsg(
                NameServerEventCode.PULL_BROKER_IP_LIST.getCode(),
                JSON.toJSONBytes(pullBrokerIpDTO)
        );
        // 发送
        TcpMsg resp = nameServerNettyRemoteClient.sendSyncMsg(
                req,
                fetchBrokerAddressMsgId
        );

        //获取broker节点ip地址，并且缓存起来，可能由多个master-broker角色
        PullBrokerIpRespDTO pullBrokerIpRespDTO = JSON.parseObject(resp.getBody(), PullBrokerIpRespDTO.class);
        setBrokerAddressList(pullBrokerIpRespDTO.getAddressList());
        log.info("拉取broker节点ip地址成功:{}", pullBrokerIpRespDTO);
    }

    /**
     * 连接到节点
     */
    private void connectBroker() {
        AssertUtils.isNotEmpty(this.getBrokerAddressList(),"broker地址不能为空");
        for (String brokerIp : brokerAddressList) {
            String[] brokerAddressArr = brokerIp.split(":");
            BrokerNettyRemoteClient brokerNettyRemoteClient = new BrokerNettyRemoteClient(brokerAddressArr[0],
                    Integer.valueOf(brokerAddressArr[1]));
            brokerNettyRemoteClient.buildConnection();
            this.brokerNettyRemoteClientMap.put(brokerIp,brokerNettyRemoteClient);
        }
    }

    @Override
    public SendResult send(MessageDTO messageDTO) {
        // todo 路由关系, 通过topic定位到broker实例(定位那个topic存储在那个broker实例中)
        //路由的关系(topic->定位到具体的broker实例的ip)
        BrokerNettyRemoteClient remoteClient = getRemoteClient(messageDTO.getTopic());
        String msgId = UUID.randomUUID().toString();
        // 发送消息
        messageDTO.setMsgId(msgId);
        messageDTO.setSendWay(BrokerMessageSendWay.SYNC.getCode());
        byte[] jsonBytes = JSON.toJSONBytes(messageDTO);
        log.info("发送消息体:{}", Arrays.toString(jsonBytes));
        TcpMsg requestMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(), jsonBytes);

        log.info("推送消息，消息ID:{}", msgId);
        TcpMsg responseMsg = remoteClient.sendSyncMsg(requestMsg, msgId);

        ClientSendMessageToBrokerRespDTO responseDTO =
                JSON.parseObject(responseMsg.getBody(), ClientSendMessageToBrokerRespDTO.class);
        int responseStatus = responseDTO.getStatus();
        // 转换成SendResult
        SendResult sendResult = new SendResult();
        if(responseStatus == ClientSendMessageToBrokerRespStatus.SUCCESS.getCode()) {
            sendResult.setSendStatus(SendStatus.SUCCESS);
        } else if (responseStatus == ClientSendMessageToBrokerRespStatus.FAIL.getCode()) {
            sendResult.setSendStatus(SendStatus.FAIL);
        }
        return sendResult;
    }



    @Override
    public void sendAsync(MessageDTO messageDTO) {
        BrokerNettyRemoteClient remoteClient = getRemoteClient(messageDTO.getTopic());;
        messageDTO.setSendWay(BrokerMessageSendWay.ASYNC.getCode());
        TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.PUSH_MSG.getCode(),JSON.toJSONBytes(messageDTO));
        remoteClient.sendAsyncMsg(tcpMsg);
    }

    /**
     * 获取broker实例
     * @param topic  topic
     * @return broker实例
     */
    private BrokerNettyRemoteClient getRemoteClient(String topic) {

        return this.brokerNettyRemoteClientMap.values().stream().collect(Collectors.toList()).get(0);
    }
}
